package com.xncoding.rocketmq.producer;

import com.xncoding.rocketmq.basic.Order;
import com.xncoding.rocketmq.producer.callback.AsyncSendCallback;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/***
* @description 消息发送端封装
* @author luming
* @date 2021/4/24 15:51
* @throws
*/
@Slf4j
@Component
public class SpringProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送同步消息
     * @param topic
     * @param msg
     */
    public void sendSyncMessage(String topic,String msg){
        SendResult sendResult = rocketMQTemplate.syncSend(topic,msg);
        log.info("发送同步消息，回执信息：{}",sendResult);
    }

    /**
     * 发送异步消息
     * @param topic
     * @param msg
     */
    public void sendAsyncMessage(String topic,String msg){
        rocketMQTemplate.asyncSend(topic,msg,new AsyncSendCallback());
    }

    /**
     * 发送同步的顺序消息，保证一类消息发送到同一个MessageQueue中
     * RocketMqTemplate底层默认使用了{@link org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash}
     * 传过去的hashKey就是arg
     * @param topic
     * @param order
     */
    public void sendOrdelyMessage(String topic, Order order){
        Message<Order> message = MessageBuilder.withPayload(order)
                .setHeader(RocketMQHeaders.KEYS,order.getOrderId()).build();
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic,message,order.getOrderId().toString());
        log.info("发送同步的顺序消息，回执信息：{}",sendResult);
    }

    /**
     * 延迟消息发送
     * messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * @param topic
     * @param msg
     */
    public void sendDelayMessage(String topic, String msg){
        Message<String> message = MessageBuilder.withPayload(msg).build();
        SendResult sendResult = rocketMQTemplate.syncSend(topic,message,2000L,4);
        log.info("发送同步的延迟消息，回执信息：{}",sendResult);
    }

    /**
     * 发送批量消息
     * RocketMqTemplate底层会对消息切割，不须要考虑超过1M的情况
     * @param topic
     * @param msgs
     */
    public void sendBatchMessage(String topic, String... msgs){
        List<Message<String>> messageList = Arrays.stream(msgs)
                .map(msg -> MessageBuilder.withPayload(msg).build())
                .collect(Collectors.toList());
        SendResult sendResult = rocketMQTemplate.syncSend(topic,messageList);
        log.info("发送同步的批量消息，回执信息：{}",sendResult);
    }

    public void sendFilterMessage(String topic,String tag, String msg){
        Message<String> message = MessageBuilder.withPayload(String.format("%s:%s",tag,msg)).build();
        String destination = String.format("%s:%s",topic,tag);
        SendResult sendResult = rocketMQTemplate.syncSend(destination,message);
        log.info("发送同步的过滤消息，回执信息：{}",sendResult);
    }


    public void sendMessageInTransaction(String topic,String msg) throws InterruptedException {
        String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            //尝试在Header中加入一些自定义的属性。
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader(RocketMQHeaders.TRANSACTION_ID,"TransID_"+i)
                    //发到事务监听器里后，这个自己设定的TAGS属性会丢失。但是上面那个属性不会丢失。
                    .setHeader(RocketMQHeaders.TAGS,tags[i % tags.length])
                    //MyProp在事务监听器里也能拿到，为什么就单单这个RocketMQHeaders.TAGS拿不到？这只能去调源码了。
                    .setHeader("MyProp","MyProp_"+i)
                    .build();
            String destination =topic+":"+tags[i % tags.length];
            //这里发送事务消息时，还是会转换成RocketMQ的Message对象，再调用RocketMQ的API完成事务消息机制。
            SendResult sendResult = rocketMQTemplate.sendMessageInTransaction(destination, message,destination);
            System.out.printf("%s%n", sendResult);

            Thread.sleep(10);
        }
    }
}
